package com.csw.flink.sql

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api._
import org.apache.flink.table.api.bridge.scala._
import org.apache.flink.types.Row

object Demo04EventTime {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val settings: EnvironmentSettings = EnvironmentSettings.newInstance()
      .useBlinkPlanner()
      .inStreamingMode()
      .build()


    val bsEnv: StreamTableEnvironment = StreamTableEnvironment.create(env, settings)


    /*
    数据为：
    csw,20200603202130
    csw,20200603202131
    csw,20200603202132
    csw,20200603202133
    csw,20200603202134
    csw,20200603202135
    csw,20200603202136
    csw,20200603202137
    csw,20200603202138
    csw,20200603202139
    csw,20200603202140
    csw,20200603202141
    csw,20200603202142
    csw,20200603202143
     */

    bsEnv.executeSql(
      """
        |
        |CREATE TABLE eventTime (
        |  word STRING,
        |  user_action_time STRING,
        |  ts as TO_TIMESTAMP(user_action_time,'yyyyMMddHHmmss'),
        |  -- 声明 user_action_time 是事件时间属性，并且用 延迟 5 秒的策略来生成 watermark
        |WATERMARK FOR ts AS ts - INTERVAL '5' SECOND
        |) WITH (
        | 'connector' = 'kafka',
        | 'topic' = 'test_topic4',
        | 'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092',
        | 'properties.group.id' = 'csw',
        | 'format' = 'csv',
        | 'csv.ignore-parse-errors' = 'true',
        | 'scan.startup.mode' = 'earliest-offset'
        |)
        |
      """.stripMargin)

    bsEnv.from("eventTime")

    bsEnv.sqlQuery(
      """
        |
        |select
        |word,
        |TUMBLE_END(ts, INTERVAL '5' SECOND) as end_time,
        |count(1) as c
        |from eventTime
        |group by word,TUMBLE(ts, INTERVAL '5' SECOND)
        |
      """.stripMargin).toAppendStream[Row].print()

    env.execute()
  }
}
